package com.bw.ads;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import com.bw.Test.DwsTradeOrgOrderDayBean;
import com.bw.bean.A;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

public class Test9 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk9", "test4" + System.currentTimeMillis()));

        //筛选订单数据
        SingleOutputStreamOperator<String> orderInfoDS = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String tableName = jsonObject.getString("table");
                if ("order_info".equals(tableName)) {
                    return true;
                }
                return false;
            }
        });

        SingleOutputStreamOperator<JSONObject> mapStream = orderInfoDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSON.parseObject(s);
            }
        });

        SingleOutputStreamOperator<JSONObject> watermarkStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getJSONObject("data").getLong("create_time") / 1000;
            }
        }));
        watermarkStream.print();
        watermarkStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10))).process(new ProcessAllWindowFunction<JSONObject, A, TimeWindow>() {
            @Override
            public void process(ProcessAllWindowFunction<JSONObject, A, TimeWindow>.Context context, Iterable<JSONObject> iterable, Collector<A> collector) throws Exception {
                int i = 0;
                int unpay = 0;
                int unsort = 0;
                int uncommit = 0;
                int untake = 0;
                Iterator<JSONObject> iterator = iterable.iterator();
                while (iterator.hasNext()){
                    i++;
                    JSONObject next = iterator.next();
                    String status = next.getJSONObject("data").getString("status");
                    if("60010".equals(status)){
                        unpay++;
                    }else if("60040".equals(status)){
                        unsort++;
                    }else if("60050".equals(status)){
                        uncommit++;
                    }else if("60020".equals(status)){
                        untake++;
                    }
                }
                System.out.println("i = " + i);
                System.out.println("unpay = " + unpay);
                System.out.println("unsort = " + unsort);
                System.out.println("uncommit = " + uncommit);
                System.out.println("untake = " + untake);
                System.out.println(unpay/i);
            }
        });

        env.execute();


    }
}
